package com.aididi.consumer;

import com.aididi.constant.Constants;
import com.aididi.entity.Order;
import com.aididi.util.JsonUtil;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Collections;
import java.util.Map;

/**
 * @Auther: xuzexiang
 * @Date: 2018-11-17 0017 21:51
 * @Description:    订单队列----》消息接收
 *
 *  * 在以下3种情况会进入死信队列
 * 1.有消息被拒绝（basic.reject/ basic.nack）并且requeue=false
 * 2.队列达到最大长度
 * 3.消息TTL过期
 * arguments        键	值	意义
 * x-message-ttl	数字类型，标志时间，以豪秒为单位	标志队列中的消息存活时间，也就是说队列中的消息超过了制定时间会被删除
 * x-expires	数字类型，标志时间，以豪秒为单位	队列自身的空闲存活时间，当前的queue在指定的时间内，没有consumer、basic.get也就是未被访问，就会被删除。
 * x-max-length和x-max-length-bytes	数字	最大长度和最大占用空间，设置了最大长度的队列，在超过了最大长度后进行插入会删除之前插入的消息为本次的留出空间,相应的最大占用大小也是这个道理，当超过了这个大小的时候，会删除之前插入的消息为本次的留出空间。
 * x-dead-letter-exchange和x-dead-letter-routing-key	字符串	消息因为超时或超过限制在队列里消失，这样我们就丢失了一些消息，也许里面就有一些是我们做需要获知的。而rabbitmq的死信功能则为我们带来了解决方案。设置了dead letter exchange与dead letter routingkey（要么都设定，要么都不设定）那些因为超时或超出限制而被删除的消息会被推动到我们设置的exchange中，再根据routingkey推到queue中
 * x-max-priority	数字	队列所支持的优先级别，列如设置为5，表示队列支持0到5六个优先级别，5最高，0最低，当然这需要生产者在发送消息时指定消息的优先级别，消息按照优先级别从高到低的顺序分发给消费者
 * alternate-exchange		下面简称AE，当一个消息不能被route的时候，如果exchange设定了AE，则消息会被投递到AE。如果存在AE链，则会按此继续投递，直到消息被route或AE链结束或遇到已经尝试route过消息的AE。
 *
 *延时队列三种方式：
 * 1.设置队列的x-message-ttl 属性de
 * 2.在生产者方设置消息的过期时间
 * 3.利用rabbitmq的rabbitmq_delayed_message_exchange插件
 * 但这两种方式是有区别的，如果设置了队列的TTL属性，那么一旦消息过期，就会被队列丢弃，而第二种方式，消息即使过期，也不一定会被马上丢弃，因为消息是否过期是在即将投递到消费者之前判定的，如果当前队列有严重的消息积压情况，则已过期的消息也许还能存活较长时间。
 * 比如同时有两条延迟消息，第一条设置10s过期，第二条设置2s过期，因为RabbitMQ只会检查第一个消息是否过期，如果过期则丢到死信队列，如果第一个消息的延时时长很长，而第二个消息的延时时长很短，则第二个消息并不会优先得到执行。
 */
@Component
@Slf4j
public class WorkConsumer {
    /**
     *	@Payload(消息体)	    @Headers(消息头)
     */
    @RabbitListener(bindings = {@QueueBinding(
            value = @Queue(value = "order-queue",//队列名称
                    durable = "true",
                    arguments = {//  将普通队列绑定到死信队列交换机上
                            //之前的队列没有绑定死信队列和死信交换机 不能做更改绑定死信交互机
                            //之前创建好的邮件队列 删除掉  已经创建好的队列不能做更改  交换机也清理掉
                            @Argument(name = "x-dead-letter-exchange", value = "order_dead_exchange"), //死信交换机
                            @Argument(name = "x-dead-letter-routing-key", value = "order_dead_routing_key.a")
                     //     @Argument(name = "x-message-ttl", value = "10000",type = "java.lang.Integer")//延时队列
                    //      @Argument(name = "x-message-ttl", value = "10000",type = "java.lang.Integer") //队列最大长度
            }),//是否持久化
            exchange = @Exchange(value = "order-exchange", //交换机名称
                    durable = "true",//是否持久化
                    type = ExchangeTypes.TOPIC, //消息路由规则
                    ignoreDeclarationExceptions = Constants.IGNOREDECEXCEPTION), // 忽略声明异常
            key = "order.#" // 绑定的路由键或模式。“*”用于匹配一个单词，“#”用于匹配多个单词
    )})
    @RabbitHandler
    public void  receive(@Payload String msg,
                         @Headers Map<String,Object> map,
                         Channel channel) throws Exception {
        Order order = JsonUtil.covertJsonToObject(msg, Order.class);
        log.info("-----接收消息-----");
        log.info("订单id"+ order.getId());
        //消息唯一标识 ID   DELIVERY_TAG， 是一个单调递增的正整数，delivery tag 的范围仅限于 Channel
        Long tag = (Long) map.get(AmqpHeaders.DELIVERY_TAG);
        try {

            /**
             * 消费端如何限流
             * 当海量消息瞬间推送过来，单个客户端无法同时处理那么多数据，严重会导致系统宕机。这时，需要削峰。
             * RabbitMQ提供了一种qos(服务质量保证)功能。即在非自动确认消息的前提下(非ACK)，
             * 如果一定数目的消息(通过基于consume或者channel设置qos的值)未被确认前，不进行消费新的消息
             *basicQos方法参数：
             * prefetchSize：0，单条消息大小限制，0代表不限制
             * prefetchCount：一次性消费的消息数量。会告诉 RabbitMQ 不要同时给一个消费者推送多于 N 个消息，即一旦有 N 个消息还没有 ack，则该 consumer 将 block 掉，直到有消息 ack。
             * global：true、false 是否将上面设置应用于 channel，简单点说，就是上面限制是 channel 级别的还是 consumer 级别。当我们设置为 false 的时候生效，设置为 true 的时候没有了限流功能，因为 channel 级别尚未实现。
             * 注意：prefetchSize 和 global 这两项，rabbitmq 没有实现，暂且不研究。特别注意一点，prefetchCount 在 no_ask=false 的情况下才生效，即在自动应答的情况下这两个值是不生效的。
             *
             */
            //channel.basicQos(0,1,false);
            /**
             * 消费端消息幂等性保障
             *  1.存redis
             *  2.通过消息唯一id查询
             */
            //1.先查询redis是否存在消息，存在则该消息已被消费过,直接return
            //单消费者例子从string取
            //以消息唯一 id为 key，订单状态为value，存入redis
            /*String messageRedisValue = redisUtil.get(order.id());
            if (messageRedisValue.equals(order.id())) {
                return;
            }*/
            //多消费者例子从list取
            /*List<String> messageRedisValue = redisUtil.lrange("queueName4");
            if (messageRedisValue.contains(order.getId())) {
                return;
            }*/

            //2.由于生产者在发送时生成的唯一id，可通过消息唯一id查询队列里是否存在这条消息，如果不存在，证明消息已被消费过了，队列此时已移除了该消息，直接return
            //可能存在高并发问题，这里不太推荐
            /* if(StringUtils.isEmpty(order.getId())){
                return;
            }*/

            //将id存入string中(单消费者场景)：
            //redisUtil.set("queueName4",order.getId());//以队列为key，id为value
            //将id存入list中(多消费者场景)：
            //redisUtil.lpush("queueName4",order.getId());//存入list
            /**
             *  手工ack认证
             *  第一个参数是唯一标识id
             *  第二个是  是否批量.true,  为了减少网络流量，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息
             */
            //这段代码会抛异常，目的是测试消息是否会进死信队列
            System.out.println(1 / 0);
            channel.basicAck(tag,false);
        } catch (Exception e){
                //拒绝消费消息（丢失消息） 给死信队列
                //消息被(basic.reject() or basic.nack()) and requeue = false，即消息被消费者拒绝或者nack，并且重新入队为false。
                //nack()与reject()的区别是：reject()不支持批量拒绝，而nack()可以.
                /**
                 * 参数说明：
                 * deliveryTag:该消息的index
                 * multiple：是否批量.true:将一次性拒绝所有小于deliveryTag的消息。
                 * requeue：被拒绝的是否重新入队列
                 */
                channel.basicNack(tag, false, false);
                log.error("订单消费者 消费异常:{},{}", order, e.getMessage());
                //重新分配(消费者会一直消费这条消息，直到消费成功)
                // 异常消费，将消息重新放入队列里
                //   channel.basicNack(tag,false,true);

        }
    }


}
